package com.yihaodian.search;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import com.yihaodian.api.common.Response;
import com.yihaodian.api.search.SearchException;
import com.yihaodian.api.search.SearchRequest;
import com.yihaodian.api.search.SearchService;
import com.yihaodian.cluster.ClusterService;
import com.yihaodian.search.broker.OperationRouting;
import com.yihaodian.search.broker.ThreadPools;
import com.yihaodian.search.broker.shard.ShardSearchClient;
import com.yihaodian.search.broker.shard.ShardSearchRequest;

/**
 * TODO: A super abstract class to handle all distribute process. Use ThreadPool
 * to handle asynchronous request, and Listener to process the requests.
 * 
 * @author zhaowei1
 * 
 */
public abstract class SearchPhaseBroker implements SearchService {
	protected ClusterService clusterService;
	protected ShardSearchClient client;
	protected OperationRouting operationRouting;
	// TODO: Cache need to be global.
	protected final SearchCache searchCache = new SearchCache();

	public SearchPhaseBroker(ClusterService clusterService,
			ShardSearchClient client) {
		this.clusterService = clusterService;
		this.operationRouting = new OperationRouting();
		this.client = client;
	}

	protected abstract void doExecute(SearchRequest request,
			SearchListener<Response> listener);

	protected abstract class BaseAsyncManager<FirstResult extends SearchPhaseResult> {
		protected SearchRequest request;
		protected SearchListener<Response> listener;

		AtomicInteger requestCounter;

		public BaseAsyncManager(SearchRequest request,
				SearchListener<Response> listener) {
			this.request = request;
			this.listener = listener;
		}

		public void start() {
//			List<ShardSearchRequest> shardRequests = operationRouting
//					.shardRequest(clusterService.state(), request);
			List<ShardSearchRequest> shardRequests = null;
			requestCounter = new AtomicInteger(shardRequests.size());
			for (final ShardSearchRequest shardRequest : shardRequests) {
				ThreadPools.getThreadPool("search").execute(new Runnable() {
					@Override
					public void run() {
						performFirstPhase(shardRequest);
					}
				});

			}
		}

		void performFirstPhase(ShardSearchRequest shardRequest) {
			FirstResult result = sendExecuteFirstPhase(shardRequest);
			processFirstPhaseResult(result);
			if (requestCounter.decrementAndGet() == 0) {
				moveToSecondPhase();
			}
		}

		// void onFirstResult(FirstResult result){
		//
		// }

		protected abstract FirstResult sendExecuteFirstPhase(
				ShardSearchRequest shardRequest);

		// TODO : Async to call the search service.
		// protected abstract void sendExecuteFirstPhase(ShardSearchRequest
		// shardRequest, SeacherServiceListener li);
		protected abstract void processFirstPhaseResult(FirstResult result);

		protected abstract void moveToSecondPhase() throws SearchException;
	}

	/**
	 * Use thread to invoke the callback method.
	 * 
	 * @author zhaowei1
	 * 
	 * @param <Response>
	 */
	protected static class ThreadedListener<Response> implements
			SearchListener<Response> {
		private final SearchListener<Response> listener;

		public ThreadedListener(SearchListener<Response> listener) {
			this.listener = listener;
		}

		@Override
		public void onResponse(final Response response) {
			ThreadPools.getThreadPool("search").execute(new Runnable() {
				@Override
				public void run() {
					try {
						listener.onResponse(response);
					} catch (Exception e) {
						listener.onFailure(e);
					}
				}
			});
		}

		@Override
		public void onFailure(final Throwable e) {
			ThreadPools.getThreadPool("search").execute(new Runnable() {
				@Override
				public void run() {
					listener.onFailure(e);
				}
			});
		}
	}
}
